package com.gitee.xhs.inbuiltWatermarkGenerator;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.WatermarkRecordMakerProcessFunction;
import com.gitee.xhs.WatermarkSourceFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static com.gitee.xhs.WatermarkConstant.DEFAULT_WATERMARK_STRATEGY;
import static com.gitee.xhs.WatermarkConstant.*;

/**
 * @author BigHead
 * @date 2022/7/18 10:23
 * @description MonotonousTimestampsMain
 *
 * monotonous 就是实现 BoundedOutOfOrderness 使用 Duration.ofMillis(0)
 *
 */
public class MonotonousTimestampsMain {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.webUi(new EnvironmentConfig(null));

        streamExecutionEnvironment
                .addSource(
                        WatermarkSourceFactory.defaultTransactionBoundOneSecondSource()
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .forMonotonousTimestamps()
                )
                .process(
                        new WatermarkRecordMakerProcessFunction(DEFAULT_WATERMARK_STRATEGY, INBUILT_MONOTONOUS_TIMESTAMPS, RECORD_TIMESTAMP_ASSIGNER)
                )
                .print();

        streamExecutionEnvironment.execute("MonotonousTimestampsMain");
    }

}
